package com.zyn.es.basic;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkProcessor.Builder;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;

/**
 * 带监控的批量操作
 * 
 * https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/java-docs-bulk-processor.html
 * @author tongyufu
 *
 */
public class BulkProcessorAPI extends EsClient {

    public static void main(String[] args) throws Exception {
        new BulkProcessorAPI().bulkProcessor();
    }

    public void bulkProcessor() throws Exception {
        super.initClient();
        log.info("================================");
        Builder build = BulkProcessor.builder(client, new BulkProcessor.Listener() {

            @Override
            public void beforeBulk(long executionId, BulkRequest request) {
                log.info("有效请求数量：" + request.numberOfActions());
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                log.info("执行出错：" + failure.getMessage());
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                log.info("执行结束：" + response.hasFailures());
                for (BulkItemResponse item : response) {
                    if (item.isFailed()) {
                        log.info(item.getFailureMessage());
                    }
                }
            }
        });
        //每次处理数量，默认1000
        build.setBulkActions(1000);
        //每10MB flush一次，默认5MB
        build.setBulkSize(new ByteSizeValue(10, ByteSizeUnit.MB));
        //不管有多少请求，5秒flush一次
        build.setFlushInterval(TimeValue.timeValueSeconds(5));
        //并发请求数量，0表示单线程
        build.setConcurrentRequests(1);
        //指定100毫秒后开始获取返回信息，最多重试3次，如果出错会抛出EsRejectedExecutionException，这表明没有足够的计算机资源来处理请求。
        //禁用重试：BackoffPolicy.noBackoff()
        build.setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3));

        BulkProcessor bulk = build.build();
        Map<String, Object> json = new HashMap<String, Object>();
        json.put("title", "BulkProcessor 10");
        bulk.add(new IndexRequest(INDEX, TYPE, "20").source(json));
        bulk.add(new DeleteRequest(INDEX, TYPE, "20"));

        //指定时间后关闭连接。如果在指定时间前所有请求处理完毕，返回true，反之返回false。
        //close()方法则不等待任何请求完成而立刻退出
        boolean result = bulk.awaitClose(10, TimeUnit.MINUTES);
        log.info("====close：" + result);
    }

}
